package com.example.rabbitmq.ksdemo.send;

import com.example.rabbitmq.bo.MqConfig;
import com.example.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.SneakyThrows;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

public class FanoutSend {

    private static Map<String, Channel> channelMap = new HashMap();

    public static void main(String[] args) {

        Thread thread=new Thread(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                MqConfig mqConfig=new MqConfig();
                mqConfig.setExName("FANOUT_EX");
                mqConfig.setQueueName(Arrays.asList("FANOUT_QUEUE_1","FANOUT_QUEUE_2"));
                mqConfig.setRouting(Arrays.asList("MD_BOND","MD_XSWAP","MD_REQ"));
                while (true){
                    for (int i =1;i<=400000;i++){
                        String key = mqConfig.getRouting().get(i % 3);
                        String message = key+"====="+i;
                        // 5. 向指定的队列中发送消息    参数2：   没有创建交换机  routingKey==队列的名字
                        sendMsg(message,mqConfig, key);
                        Thread.sleep(1500);
                    }
                }
            }
        });
        thread.start();

    }

    public static void sendMsg(String payload , MqConfig mqConfig,String key) throws IOException, TimeoutException {
        String exName = mqConfig.getExName();
        Channel channel = getChannel(mqConfig, exName);
        channel.basicPublish(exName, key, (AMQP.BasicProperties)null, payload.getBytes("UTF-8"));
    }

    private static Channel getChannel(MqConfig mqConfig, String exName) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = channelMap.get(exName);

        if (channel == null) {
            channel = connection.createChannel();
            channel.exchangeDeclare(exName, BuiltinExchangeType.FANOUT);
//            channel.exchangeDeclare(exName, BuiltinExchangeType.TOPIC);
            Map<String, Object> args = new HashMap();
            args.put("x-queue-type", "classic");
//            channel.queueDeclare(mqConfig.getQueueName().get(0),true,false,false,args);
//            channel.queueDeclare(mqConfig.getQueueName().get(1),true,false,false,args);
//            channel.queueBind(mqConfig.getQueueName().get(0),exName,"");
//            channel.queueBind(mqConfig.getQueueName().get(1),exName,"");
            channelMap.put(exName, channel);
            System.out.println("3================:" + channel);
            return channel;
        } else {
            System.out.println("4==================:" + channel);
            if (channel.isOpen()) {
                return channel;
            } else {
                System.out.println("5================:" + channel);
//                safeClose(channel);
                channel = connection.createChannel();
                channelMap.put(exName, channel);
                System.out.println("6:" + channel);
                return channel;
            }
        }
    }
}
